package org.xi.maple.source

import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.slf4j.{Logger, LoggerFactory}
import org.xi.maple.api.MapleSource

import scala.collection.mutable

class Jdbc extends MapleSource[JdbcConfig] {

  private val log: Logger = LoggerFactory.getLogger(classOf[Jdbc])

  override def checkConfig: (Boolean, String) = {
    val nonExistsOptions = mutable.MutableList[String]()
    if (StringUtils.isBlank(config.getUrl)) {
      nonExistsOptions.+=:("url")
    }
    if (StringUtils.isBlank(config.getDriver)) {
      nonExistsOptions.+=:("driver")
    }
    if (StringUtils.isBlank(config.getQuery)) {
      nonExistsOptions.+=:("query")
    }
    if (StringUtils.isBlank(config.getUser)) {
      nonExistsOptions.+=:("user")
    }
    if (StringUtils.isBlank(config.getResultTable)) {
      nonExistsOptions.+=:("resultTable")
    }

    if (nonExistsOptions.isEmpty) {
      (true, "")
    } else {
      (false, "please specify " + nonExistsOptions.map { field => s"[${field}]" }.mkString(", ") + " as non-empty string")
    }
  }

  override def getData(spark: SparkSession, argsMap: mutable.Map[String, String]): Dataset[Row] = {
    val reader = spark.read.format("jdbc")
    if (config.getOptions != null && !config.getOptions.isEmpty) {
      reader.options(config.getOptions)
    }

    log.info(s"Load data from jdbc url: ${config.getUrl}, driver: ${config.getDriver}, username: ${config.getUser}, query: ${config.getQuery}")

    val result = reader
      .option("url", config.getUrl)
      .option("driver", config.getDriver)
      .option("query", config.getQuery)
      .option("user", config.getUser)
      .option("password", config.getPassword).load()

    // todo 待优化
    if (config.getCache) {
      result.persist()
    }
    result
  }
}
